package baiduschPro;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.stream.io.ByteArrayOutputStream;

import hylanda.common.save2MysqlBase;


@Tags({"update mysql"})
@CapabilityDescription("update mysql")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})
public class UpdateMySql extends AbstractProcessor {

	 public static final PropertyDescriptor MYSQLURL = new PropertyDescriptor
	            .Builder().name("MYSQLURL")
	            .displayName("mysqlurl")
	            .description("mysql url")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYSQLUSER = new PropertyDescriptor
	            .Builder().name("MYSQLUSER")
	            .displayName("mysqluser")
	            .description("mysql user")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYSQLPWD = new PropertyDescriptor
	            .Builder().name("MYSQLPWD")
	            .displayName("mysqlpwd")
	            .description("mysql password")
	            .required(false)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	
	 public static final PropertyDescriptor MYSQLTBNAME = new PropertyDescriptor
	            .Builder().name("TBNAME")
	            .displayName("table name")
	            .description("table name")
	            .required(false)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYFIELDLIST = new PropertyDescriptor
	            .Builder().name("MYFIELDLIST")
	            .displayName("update field list")
	            .description("要存储的字段名称列表，逗号分隔")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYFIELDTYPELIST = new PropertyDescriptor
	            .Builder().name("MYFIELDTYPELIST")
	            .displayName("update field type list")
	            .description("要存储的字段名称列表，逗号分隔")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYFIELDWHEREFIELD = new PropertyDescriptor
	            .Builder().name("MYFIELDWHEREFIELD")
	            .displayName("where field")
	            .description("要更新的wher中条件字段，逗号分隔")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYFIELDWHEREFIELDTYPE = new PropertyDescriptor
	            .Builder().name("MYFIELDWHEREFIELDTYPE")
	            .displayName("where field type")
	            .description("要更新的wher中条件字段类型，逗号分隔")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 
	 public static final Relationship MY_RELATIONSHIPOK = new Relationship.Builder()
	            .name("gen success")
	            .description("success relationship")
	            .build();
	    
	    public static final Relationship MY_RELATIONSHIPNO = new Relationship.Builder()
	            .name("gen failure")
	            .description("failure relationship")
	            .build();

	    private save2MysqlBase mysqlselect = new save2MysqlBase();
	    private List<PropertyDescriptor> descriptors;
	    private Set<Relationship> relationships;
	    
	 @Override
	 protected void init(final ProcessorInitializationContext context) {
	     final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
	     descriptors.add(MYSQLURL);
	     descriptors.add(MYSQLUSER);
	     descriptors.add(MYSQLPWD);
	     descriptors.add(MYSQLTBNAME);
	     descriptors.add(MYFIELDLIST);
	     descriptors.add(MYFIELDTYPELIST);
	     descriptors.add(MYFIELDWHEREFIELD);
	     descriptors.add(MYFIELDWHEREFIELDTYPE);
	     this.descriptors = Collections.unmodifiableList(descriptors);
	
	     final Set<Relationship> relationships = new HashSet<Relationship>();
	     relationships.add(MY_RELATIONSHIPOK);
	     relationships.add(MY_RELATIONSHIPNO);
	     this.relationships = Collections.unmodifiableSet(relationships);
	 }
	
	 @Override
	 public Set<Relationship> getRelationships() {
	     return this.relationships;
	 }
	
	 @Override
	 public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
	     return descriptors;
	 }
	public UpdateMySql() {
		// TODO Auto-generated constructor stub
	}

	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		// TODO Auto-generated method stub
		
		FlowFile flowFile = session.get();
	      if ( flowFile == null ) 
	      {
	      	
	      	System.out.println("flowfile is null");
	         return;
	      }
	      
	      //读取FlowFile
	        String content = "";
	        InputStream in = session.read(flowFile);
	        try {
				byte[] by = readStream(in);
				content = new String(by,"UTF-8");
				
			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
				//return ;
			}
	        
	        //获得propty
	        PropertyValue mysqlurlp = context.getProperty("MYSQLURL");
	    	PropertyValue mysqluserp= context.getProperty("MYSQLUSER");
	    	PropertyValue mysqlpwdp = context.getProperty("MYSQLPWD");
	    	PropertyValue mysqltbnamep = context.getProperty("TBNAME");
	    	PropertyValue fieldlistp = context.getProperty("MYFIELDLIST");
	    	PropertyValue wherefieldp = context.getProperty("MYFIELDWHEREFIELD");
	    	PropertyValue wherefieldtypep = context.getProperty("MYFIELDWHEREFIELDTYPE");
	    	PropertyValue fieldtypelistp = context.getProperty("MYFIELDTYPELIST");
	    	
	       String mysqlurl =  mysqlurlp.getValue();
	       String mysqluser = mysqluserp.getValue();
	       String mysqlpwd = mysqlpwdp.getValue();
	       String tbname = mysqltbnamep.getValue();
	       String fieldlist =  fieldlistp.getValue();
	       String wherefield =  wherefieldp.getValue();
	       String fieldtypelist = fieldtypelistp.getValue();
	       String wherefieldtype= wherefieldtypep.getValue();
	       
	       mysqlselect.setParams(mysqlurl,mysqluser,mysqlpwd);
	       mysqlselect.initmysql();
	       int ncnt = mysqlselect.updateDatas(content, tbname, fieldlist, fieldtypelist,wherefield,wherefieldtype);
	       mysqlselect.deconnSQL();
	       
	       ///
	     //写flowfile
	     final String strMessage = String.format("save cnt=%d", ncnt);
	     flowFile= session.write(flowFile, new OutputStreamCallback(){
	    		@Override
	    		public void process(OutputStream out) throws IOException{
	    			out.write(strMessage.getBytes("UTF-8"));
	    		}
	    	});
	     
	     session.transfer(flowFile,MY_RELATIONSHIPOK);
	
	}
	
	
	
	public static byte[] readStream(InputStream inStream) throws Exception {  
        ByteArrayOutputStream outSteam = new ByteArrayOutputStream();  
        byte[] buffer = new byte[1024];  
        int len = -1;  
        while ((len = inStream.read(buffer)) != -1) {  
            outSteam.write(buffer, 0, len);  
        }  
        outSteam.close();  
        inStream.close();  
        return outSteam.toByteArray();  
    }  

}
